package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.DimConfigBean;
import com.intct.util.HBaseUtil;
import com.intct.util.JDBCUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.sql.Connection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-08-20 17:15
 */
public class LookupSyncFunc extends RichMapFunction<JSONObject, String> {
    private Map<String, DimConfigBean> configMap = new HashMap<>();

    private org.apache.hadoop.hbase.client.Connection hbaseConnection;

    private String namespace;
    private String tableName;
    // 主流中ID字段
    private String id;


    public LookupSyncFunc(String namespace, String tableName, String id) {
        this.id = id;
        this.namespace = namespace;
        this.tableName = tableName;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        // 采用初化配置表
        Connection connection = JDBCUtil.getJDBCConnect();
        List<DimConfigBean> dimConfigBeanList =
                JDBCUtil.getQuery(connection, "select * from travel_config.dim_table_config", DimConfigBean.class);
        for (DimConfigBean dimConfigBean : dimConfigBeanList) {
            configMap.put(dimConfigBean.getSinkTable(), dimConfigBean);
        }
        if (connection != null) {
            connection.close();
        }

        hbaseConnection = HBaseUtil.getConnection();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(hbaseConnection);
    }

    @Override
    public String map(JSONObject jsonObj) throws Exception {

        // ID
        String driverId = jsonObj.getString(id);

        //
        DimConfigBean dimConfigBean = configMap.get(tableName);

        // 查询HBase维度表 {namespace table_name rowkey family columns}
        String dimDriverInfo = HBaseUtil.getRowkey(hbaseConnection, namespace, tableName, driverId);
        JSONObject driverJsonObj = JSON.parseObject(dimDriverInfo);
        // 将维护中rowkey字段去掉
        driverJsonObj.remove(dimConfigBean.getSinkRowKey());

        // 将维护字段加入到主流
        jsonObj.putAll(driverJsonObj);

        return jsonObj.toJSONString();
    }
}
